#include "transmission/tcp_server.h"
#include "log.h"
#include "ret.h"
#include "message.h"

#include<stddef.h>
#include<stdlib.h>
#include<string.h>
#include<unistd.h>
#include<arpa/inet.h> 

ret_t tcp_server_notify(tcp_server_t* server)
{
    int ret = 0;
    ret = pthread_mutex_lock(&server->mutex_to_received_msg);
    if (ret != 0) {
        return PTHREAD_MUTEX_LOCK_FAILED;
    }
    ret = pthread_cond_signal(&(server->flag_to_received_msg));
    if (ret != 0) {
        LOG_ERROR("tcp pthread notify failed");
    }

    ret = pthread_mutex_unlock(&server->mutex_to_received_msg);
    if (ret != 0) {
        return PTHREAD_MUTEX_UNLOCK_FAILED;
    }

    return RET_SUCCESS;
}

ret_t tcp_server_handle_new_connection(tcp_server_t* server)
{
    int connect_td = -1;
    struct sockaddr client_addr;
    socklen_t client_addr_len = 0;
    connect_td = accept(server->server_fd, &client_addr, &client_addr_len);
    if (connect_td == -1) {
        LOG_ERROR("accept failed");
        return TCP_CONNECT_FAILED;
    }
    struct epoll_event ev; //用于注册事件
    ev.events = EPOLLIN;
    ev.data.fd = connect_td;
    int ret = epoll_ctl(server->epoll_fd, EPOLL_CTL_ADD, connect_td, &ev);
    if (ret < 0) {
        LOG_ERROR("epoll_ctl failed");
        return EPOLL_CTL_ADD_FAILED; // TODO: release resource
    }
    return RET_SUCCESS;
}

ret_t tcp_recv_message(tcp_server_t* server, int connect_fd)
{
    int bytes = 0;
    int recv_bytes = 0;
    int message_size = 0;

    bytes = read(connect_fd, &message_size, sizeof(int));
    if (bytes != sizeof(int)) {
        LOG_ERROR("read message size failed");
        return READ_SIZE_NOT_EQAL_EXPECT; // TODO: 资源回收
    }
    // 网络转本地字节序
    // 接下来需要读取多少bytes的数据
    message_size = ntohl(message_size);
    if(message_size < 0) {
        LOG_ERROR("message size is invalid");
        return READ_MESSAGE_SIZE_INVALID;
    }
    // 创建message_t保存接收到的数据
    message_t* message = message_create(message_size + sizeof(int),
        MESSAGE_PRIORITY_DEFAULT);
    if (message == NULL) {
        LOG_ERROR("create message failed");
        return MESSAGE_CREATE_FAILED;
    }
    // 读取指定长度的数据,存放到buff中
    recv_bytes = 0;
    char* buff = message->data;
    (void)memcpy(buff, &message_size, sizeof(int));
    // TODO: 这里需要对接收到的数据进行解码
    while(recv_bytes < message_size) {
        bytes = read(connect_fd, buff + 4 + recv_bytes, (message_size - recv_bytes));
        if (bytes < 0) {
            LOG_ERROR("read message failed");
            break;
        }
        recv_bytes += bytes;
    }
    // 将接收到的消息入队
    ret_t ret = message_enqueue(server->recv_queue, message);
    if (ret != RET_SUCCESS) {
        return ret;
    }

    // 通知应用层取数据
    ret = tcp_server_notify(server);
    if (ret != RET_SUCCESS) {
        return ret;
    }

    recv_bytes = htonl(recv_bytes);
    // 向客户端确认自己接收到了多少数据
    bytes = write(connect_fd, &recv_bytes, sizeof(int));
    if (bytes != sizeof(int)) {
        LOG_ERROR("write recv bytes failed");
        return WRITE_RECV_BYTES_FAILED;
    }

    return RET_SUCCESS;
}

ret_t tcp_server_start(tcp_server_t* server)
{
    // 将server_fd本身也注册进来
    struct epoll_event ev;
    ev.events = EPOLLIN;
    ev.data.fd = server->server_fd;
    epoll_ctl(server->epoll_fd, EPOLL_CTL_ADD, server->server_fd, &ev);
    
    size_t ready_num = 0;
    int i = 0;
    ret_t status = RET_SUCCESS;
    int connect_fd = -1;
   
    while (g_server_run_flag) {
        // -1: 表示没有消息进来时，一直阻塞
        ready_num = epoll_wait(server->epoll_fd, server->events, EPOLL_MAX_EVENTS, -1);
        for (i = 0; i < ready_num; i++) {
            if (server->events[i].data.fd == server->server_fd) {
                // 处理新连接
                status = tcp_server_handle_new_connection(server);
                if (status != RET_SUCCESS) {
                    LOG_ERROR("tcp_server_handle_new_connection failed");
                    continue;
                }
            } else {
                // 读取消息内容
                connect_fd = server->events[i].data.fd;
                status = tcp_recv_message(server, connect_fd);
            }
        }
    }
}

tcp_server_t* tcp_server_create(tcp_server_config_t* config)
{
    // 申请空间
    tcp_server_t* server = (tcp_server_t*)malloc(sizeof(tcp_server_t));
    if(server == NULL) {
        LOG_ERROR("malloc failed");
        return NULL;
    }
    (void)memset(server, 0, sizeof(tcp_server_t));
    // 创建epoll_fd
    server->epoll_fd = epoll_create(EPOLL_MAX_EVENTS);
    if (server->epoll_fd == -1) {
        LOG_ERROR("epoll create failed");
        goto TCP_SERVER_CREATE_FAILED;
    }
    // 创建tcp socket
    server->server_fd = socket(AF_INET, SOCK_STREAM, 0);
    if(server->server_fd == -1) {
        LOG_ERROR("socket create failed");
        return NULL;
    }
    // 初始化socket参数
    server->server_addr.sin_family = AF_INET;
    // 绑定端口
    server->server_addr.sin_port = htons(config->port_min);
    // 使用INADDR_ANY，系统自动获取本机的IP地址
    server->server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    // 设置为接口复用
    int opt=1;  //设置为接口复用
    setsockopt(server->server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

    // 将本地地址绑定到socket
    int ret = bind(server->server_fd, (struct sockaddr*)&(server->server_addr),
        sizeof(server->server_addr));
    if (ret == -1) {
        LOG_ERROR("bind local addr to socket failed");
        goto TCP_SERVER_CREATE_FAILED;
    }
    // 将socket变为被动型的，开始监听是否有客户连接
    ret = listen(server->server_fd, config->backlog);
    if (ret == -1) {
        LOG_ERROR("listen socket failed");
        goto TCP_SERVER_CREATE_FAILED;
    }
    ret = pthread_cond_init(&(server->flag_to_received_msg), NULL);
    if (ret != 0) {
        LOG_ERROR("pthread_cond_init failed");
        goto TCP_SERVER_CREATE_FAILED;
    }
    ret = pthread_mutex_init(&(server->mutex_to_received_msg), NULL);
    if (ret != 0) {
        LOG_ERROR("pthread_mutex_init failed");
        goto TCP_SERVER_CREATE_FAILED;
    }
    LOG_INFO("tcp server create success");
    return server;

TCP_SERVER_CREATE_FAILED:
    tcp_server_destroy(&server);
    return NULL;
}